/*-------------------------------------------------------------------------
 *
 * pqcomm.c
 *	  前端与后端之间的通信函数
 *
 * 这些例程处理前端和后端之间通信的底层细节。它们仅仅在通信
 * 通道中传递数据，而忽略数据的语义。
 *
 * 要发送出站消息，请使用pqformat.c中的例程在缓冲区中构建
 * 消息，然后通过一次调用pq_putmessage进行发送。
 * 没有函数可以发送原始字节或部分消息；这确保了如果由于
 * ereport(ERROR)而在消息发送过程中中止，通道不会因不完整的
 * 消息而发生堵塞。
 *
 * libpq曾一度在前端和后端之间共享，但现在
 * 后端的“backend/libpq”与“interfaces/libpq”完全分开。
 * 所剩下的唯一相似之处仅是名称，以便于不谨慎者...
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *	src/backend/libpq/pqcomm.c
 *
 *-------------------------------------------------------------------------
 */

/*------------------------
 * 接口例程
 *
 * 设置/拆除:
 *		StreamServerPort	- 打开后端服务器端口
 *		StreamConnection	- 创建与客户端的新连接
 *		StreamClose			- 关闭客户端/后端连接
 *		TouchSocketFiles	- 保护套接字文件免受/tmp清理影响
 *		pq_init			- 在后端启动时初始化libpq
 *		socket_comm_reset	- 在错误恢复期间重置libpq
 *		socket_close		- 在后端退出时关闭libpq
 *
 * 低级I/O：
 *		pq_getbytes		- 从连接中获取已知数量的字节
 *		pq_getmessage	- 从连接中获取一个带长度字的消息
 *		pq_getbyte		- 从连接中获取下一个字节
 *		pq_peekbyte		- 从连接中窥探下一个字节
 *		pq_flush		- 刷新待处理的输出
 *		pq_flush_if_writable - 如果可以在不阻塞的情况下刷新待处理的输出
 *		pq_getbyte_if_available - 如果可用，则在不阻塞的情况下获取字节
 *
 * 消息级I/O
 *		pq_putmessage	- 发送正常消息（在COPY OUT模式下被压制）
 *		pq_putmessage_noblock - 缓存正常消息（在COPY OUT中被压制）
 *
 *------------------------
 */
#include "postgres.h"

#ifdef HAVE_POLL_H
#include <poll.h>
#endif
#include <signal.h>
#include <fcntl.h>
#include <grp.h>
#include <unistd.h>
#include <sys/file.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <netdb.h>
#include <netinet/in.h>
#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif
#include <utime.h>
#ifdef _MSC_VER					/* mstcpip.h 在 mingw 上缺失 */
#include <mstcpip.h>
#endif

#include "common/ip.h"
#include "libpq/libpq.h"
#include "miscadmin.h"
#include "port/pg_bswap.h"
#include "storage/ipc.h"
#include "utils/guc.h"
#include "utils/memutils.h"

/*
 * 处理各种平台特定的 TCP keepalive 套接字
 * 选项的拼写方式。这不包括 Windows，Windows 通常会独自处理。
 */
#if defined(TCP_KEEPIDLE)
/* TCP_KEEPIDLE 是此选项在 Linux 和 *BSD 上的名称 */
#define PG_TCP_KEEPALIVE_IDLE TCP_KEEPIDLE
#define PG_TCP_KEEPALIVE_IDLE_STR "TCP_KEEPIDLE"
#elif defined(TCP_KEEPALIVE_THRESHOLD)
/* TCP_KEEPALIVE_THRESHOLD 是此选项在 Solaris >= 11 上的名称 */
#define PG_TCP_KEEPALIVE_IDLE TCP_KEEPALIVE_THRESHOLD
#define PG_TCP_KEEPALIVE_IDLE_STR "TCP_KEEPALIVE_THRESHOLD"
#elif defined(TCP_KEEPALIVE) && defined(__darwin__)
/* TCP_KEEPALIVE 是此选项在 macOS 上的名称 */
/* 注意：Solaris 有此符号，但它的含义不同 */
#define PG_TCP_KEEPALIVE_IDLE TCP_KEEPALIVE
#define PG_TCP_KEEPALIVE_IDLE_STR "TCP_KEEPALIVE"
#endif

/*
 * 配置选项
 */
int			Unix_socket_permissions;
char	   *Unix_socket_group;

/* Unix套接字文件的位置（palloc'd字符串列表） */
static List *sock_paths = NIL;

/*
 * 低级I/O的缓冲区。
 *
 * 接收缓冲区是固定大小。发送缓冲区通常为8k，但可以通过
 * pq_putmessage_noblock()增大，如果消息不适合的话。
 */

#define PQ_SEND_BUFFER_SIZE 8192
#define PQ_RECV_BUFFER_SIZE 8192

static char *PqSendBuffer;
static int	PqSendBufferSize;	/* 发送缓冲区的大小 */
static int	PqSendPointer;		/* 在PqSendBuffer中存储字节的下一个索引 */
static int	PqSendStart;		/* 在PqSendBuffer中发送字节的下一个索引 */

static char PqRecvBuffer[PQ_RECV_BUFFER_SIZE];
static int	PqRecvPointer;		/* 从PqRecvBuffer中读取字节的下一个索引 */
static int	PqRecvLength;		/* 在PqRecvBuffer中可用数据的结束 */

/*
 * 消息状态
 */
static bool PqCommBusy;			/* 正在向客户端发送数据 */
static bool PqCommReadingMsg;	/* 正在读取消息的过程中 */


/* 内部函数 */
static void fc_socket_comm_reset(void);
static void fc_socket_close(int fc_code, Datum fc_arg);
static void fc_socket_set_nonblocking(bool fc_nonblocking);
static int	fc_socket_flush(void);
static int	fc_socket_flush_if_writable(void);
static bool fc_socket_is_send_pending(void);
static int	fc_socket_putmessage(char fc_msgtype, const char *fc_s, size_t fc_len);
static void fc_socket_putmessage_noblock(char fc_msgtype, const char *fc_s, size_t fc_len);
static int	fc_internal_putbytes(const char *fc_s, size_t fc_len);
static int	fc_internal_flush(void);

#ifdef HAVE_UNIX_SOCKETS
static int	fc_Lock_AF_UNIX(const char *fc_unixSocketDir, const char *fc_unixSocketPath);
static int	fc_Setup_AF_UNIX(const char *fc_sock_path);
#endif							/* HAVE_UNIX_SOCKETS */

static const PQcommMethods PqCommSocketMethods = {
	fc_socket_comm_reset,
	fc_socket_flush,
	fc_socket_flush_if_writable,
	fc_socket_is_send_pending,
	fc_socket_putmessage,
	fc_socket_putmessage_noblock
};

const PQcommMethods *PqCommMethods = &PqCommSocketMethods;

WaitEventSet *FeBeWaitSet;


/* --------------------------------
 *		pq_init - 在后端启动时初始化libpq
 * --------------------------------
 */
void pq_init(void)
{
	int			fc_socket_pos PG_USED_FOR_ASSERTS_ONLY;
	int			fc_latch_pos PG_USED_FOR_ASSERTS_ONLY;

	/* 初始化状态变量 */
	PqSendBufferSize = PQ_SEND_BUFFER_SIZE;
	PqSendBuffer = MemoryContextAlloc(TopMemoryContext, PqSendBufferSize);
	PqSendPointer = PqSendStart = PqRecvPointer = PqRecvLength = 0;
	PqCommBusy = false;
	PqCommReadingMsg = false;

	/* 设置进程退出钩子以关闭套接字 */
	on_proc_exit(fc_socket_close, 0);

	/*
	 * 在后端（自fork后）我们以非阻塞模式操作底层套接字，并
	 * 使用闩锁实现阻塞语义（如有必要）。这允许我们提供安全的可中断读取
	 * 和写入。
	 */
#ifndef WIN32
	if (!pg_set_noblock(MyProcPort->sock))
		ereport(FATAL,
				(errmsg("could not set socket to nonblocking mode: %m")));
#endif

	FeBeWaitSet = CreateWaitEventSet(TopMemoryContext, FeBeWaitSetNEvents);
	fc_socket_pos = AddWaitEventToSet(FeBeWaitSet, WL_SOCKET_WRITEABLE,
								   MyProcPort->sock, NULL, NULL);
	fc_latch_pos = AddWaitEventToSet(FeBeWaitSet, WL_LATCH_SET, PGINVALID_SOCKET,
								  MyLatch, NULL);
	AddWaitEventToSet(FeBeWaitSet, WL_POSTMASTER_DEATH, PGINVALID_SOCKET,
					  NULL, NULL);

	/*
	 * 事件位置与我们添加它们的顺序匹配，但让我们进行理智
	 * 检查以确保。
	 */
	Assert(fc_socket_pos == FeBeWaitSetSocketPos);
	Assert(fc_latch_pos == FeBeWaitSetLatchPos);
}

/* --------------------------------
 *		socket_comm_reset - 在错误恢复期间重置libpq
 *
 * 该函数在外部空闲循环的错误恢复中被调用。它旨在帮助我们摆脱麻烦，避免在pqcomm.c例程内部调用elog()（理想情况下，这种情况不应该发生，但……）
 * --------------------------------
 */
static void fc_socket_comm_reset(void)
{
	/* 不要丢弃待处理的数据，但要重置忙碌标志 */
	PqCommBusy = false;
}

/* --------------------------------
 *		socket_close - 在后端退出时关闭libpq
 *
 * 这是BackendInitialize()期间存在的一个pg_on_exit_callback。该函数的异常信号处理限制了该回调必须在任何时刻安全运行。
 * --------------------------------
 */
static void fc_socket_close(int fc_code, Datum fc_arg)
{
	/* 在独立后端中没什么可做的，MyProcPort为NULL。 */
	if (MyProcPort != NULL)
	{
#ifdef ENABLE_GSS
		/*
		 * 关闭GSSAPI层。当中断BackendInitialize()时，此部分不做任何事情，因为pg_GSS_recvauth()首次使用“ctx”和“cred”。
		 *
		 * 注意，我们并不打算释放MyProcPort->gss，因为我们即将退出。
		 */
		if (MyProcPort->gss)
		{
			OM_uint32	fc_min_s;

			if (MyProcPort->gss->ctx != GSS_C_NO_CONTEXT)
				gss_delete_sec_context(&fc_min_s, &MyProcPort->gss->ctx, NULL);

			if (MyProcPort->gss->cred != GSS_C_NO_CREDENTIAL)
				gss_release_cred(&fc_min_s, &MyProcPort->gss->cred);
		}
#endif							/* ENABLE_GSS */

		/*
		 * 清理SSL层。其他地方没有后主进程子进程调用此函数，因此在中断BackendInitialize()时这是安全的。
		 */
		secure_close(MyProcPort);

		/*
		 * 以前我们在这里进行了显式的close()，但似乎更好在进程死之前保持套接字打开。这允许客户端执行“同步关闭”，如果他们在意的话——等待传输层报告连接关闭，这样你就可以确保后端已退出。
		 *
		 * 尽管如此，我们将sock设置为PGINVALID_SOCKET以防止进一步的I/O。
		 */
		MyProcPort->sock = PGINVALID_SOCKET;
	}
}



/*
 * 流 -- 包装Unix套接字系统调用
 *
 *
 *		流函数用于普通TCP连接协议。
 */


/*
 * StreamServerPort -- 打开一个“监听”端口以接受连接。
 *
 * family应为AF_UNIX或AF_UNSPEC；portNumber是端口号。对于AF_UNIX端口，hostName应为NULL，且必须指定unixSocketDir。对于TCP端口，hostName要么为NULL（表示所有接口），要么为要监听的接口，unixSocketDir则被忽略（可以为NULL）。
 *
 * 成功打开的套接字将添加到ListenSocket[]数组（长度为MaxListen）中第一个不是PGINVALID_SOCKET的位置。
 *
 * 返回：STATUS_OK或STATUS_ERROR
 */

int StreamServerPort(int fc_family, const char *fc_hostName, unsigned short fc_portNumber,
				 const char *fc_unixSocketDir,
				 pgsocket fc_ListenSocket[], int fc_MaxListen)
{
	pgsocket	fc_fd;
	int			fc_err;
	int			fc_maxconn;
	int			fc_ret;
	char		fc_portNumberStr[32];
	const char *fc_familyDesc;
	char		fc_familyDescBuf[64];
	const char *fc_addrDesc;
	char		fc_addrBuf[NI_MAXHOST];
	char	   *fc_service;
	struct addrinfo *fc_addrs = NULL,
			   *fc_addr;
	struct addrinfo fc_hint;
	int			fc_listen_index = 0;
	int			fc_added = 0;

#ifdef HAVE_UNIX_SOCKETS
	char		fc_unixSocketPath[MAXPGPATH];
#endif
#if !defined(WIN32) || defined(IPV6_V6ONLY)
	int			fc_one = 1;
#endif

	/* 初始化提示结构 */
	MemSet(&fc_hint, 0, sizeof(fc_hint));
	fc_hint.ai_family = fc_family;
	fc_hint.ai_flags = AI_PASSIVE;
	fc_hint.ai_socktype = SOCK_STREAM;

#ifdef HAVE_UNIX_SOCKETS
	if (fc_family == AF_UNIX)
	{
		/*
		 * 从portNumber和unixSocketDir创建unixSocketPath并锁定该文件路径
		 */
		UNIXSOCK_PATH(fc_unixSocketPath, fc_portNumber, fc_unixSocketDir);
		if (strlen(fc_unixSocketPath) >= UNIXSOCK_PATH_BUFLEN)
		{
			ereport(LOG,
					(errmsg("Unix-domain socket path \"%s\" is too long (maximum %d bytes)",
							fc_unixSocketPath,
							(int) (UNIXSOCK_PATH_BUFLEN - 1))));
			return STATUS_ERROR;
		}
		if (fc_Lock_AF_UNIX(fc_unixSocketDir, fc_unixSocketPath) != STATUS_OK)
			return STATUS_ERROR;
		fc_service = fc_unixSocketPath;
	}
	else
#endif							/* HAVE_UNIX_SOCKETS */
	{
		snprintf(fc_portNumberStr, sizeof(fc_portNumberStr), "%d", fc_portNumber);
		fc_service = fc_portNumberStr;
	}

	fc_ret = pg_getaddrinfo_all(fc_hostName, fc_service, &fc_hint, &fc_addrs);
	if (fc_ret || !fc_addrs)
	{
		if (fc_hostName)
			ereport(LOG,
					(errmsg("could not translate host name \"%s\", service \"%s\" to address: %s",
							fc_hostName, fc_service, gai_strerror(fc_ret))));
		else
			ereport(LOG,
					(errmsg("could not translate service \"%s\" to address: %s",
							fc_service, gai_strerror(fc_ret))));
		if (fc_addrs)
			pg_freeaddrinfo_all(fc_hint.ai_family, fc_addrs);
		return STATUS_ERROR;
	}

	for (fc_addr = fc_addrs; fc_addr; fc_addr = fc_addr->ai_next)
	{
		if (fc_family != AF_UNIX && fc_addr->ai_family == AF_UNIX)
		{
			/*
			 * 仅在请求了unix域套接字时才设置。此时服务/端口是不同的。
			 */
			continue;
		}

		/* 查看是否还有空间添加1个更多的套接字。 */
		for (; fc_listen_index < fc_MaxListen; fc_listen_index++)
		{
			if (fc_ListenSocket[fc_listen_index] == PGINVALID_SOCKET)
				break;
		}
		if (fc_listen_index >= fc_MaxListen)
		{
			ereport(LOG,
					(errmsg("could not bind to all requested addresses: MAXLISTEN (%d) exceeded",
							fc_MaxListen)));
			break;
		}

		/* 为日志消息设置地址族名称 */
		switch (fc_addr->ai_family)
		{
			case AF_INET:
				fc_familyDesc = _("IPv4");
				break;
#ifdef HAVE_IPV6
			case AF_INET6:
				fc_familyDesc = _("IPv6");
				break;
#endif
#ifdef HAVE_UNIX_SOCKETS
			case AF_UNIX:
				fc_familyDesc = _("Unix");
				break;
#endif
			default:
				snprintf(fc_familyDescBuf, sizeof(fc_familyDescBuf),
						 _("unrecognized address family %d"),
						 fc_addr->ai_family);
				fc_familyDesc = fc_familyDescBuf;
				break;
		}

		/* 为日志消息设置地址的文本形式 */
#ifdef HAVE_UNIX_SOCKETS
		if (fc_addr->ai_family == AF_UNIX)
			fc_addrDesc = fc_unixSocketPath;
		else
#endif
		{
			pg_getnameinfo_all((const struct sockaddr_storage *) fc_addr->ai_addr,
							   fc_addr->ai_addrlen,
							   fc_addrBuf, sizeof(fc_addrBuf),
							   NULL, 0,
							   NI_NUMERICHOST);
			fc_addrDesc = fc_addrBuf;
		}

		if ((fc_fd = socket(fc_addr->ai_family, SOCK_STREAM, 0)) == PGINVALID_SOCKET)
		{
			ereport(LOG,
					(errcode_for_socket_access(),
			/* 翻译者：第一个%s是IPv4，IPv6或Unix */
					 errmsg("could not create %s socket for address \"%s\": %m",
							fc_familyDesc, fc_addrDesc)));
			continue;
		}

#ifndef WIN32

		/*
		 * 如果没有 SO_REUSEADDR 标志，则在停止或崩溃后，无法立即启动新的 postmaster
		 * ，这会在 TCP 端口上给出“地址已在使用”错误。
		 *
		 * 但是在 win32 上，只有在设置了 SO_EXCLUSIVEADDRUSE 时，
		 * 这种行为才会发生。使用 SO_REUSEADDR，win32 允许多个服务器
		 * 监听同一地址，导致不可预测的行为。在没有任何标志的情况下，win32 的行为
		 * 与 Unix 的 SO_REUSEADDR 相同。
		 */
		if (fc_addr->ai_family != AF_UNIX)
		{
			if ((setsockopt(fc_fd, SOL_SOCKET, SO_REUSEADDR,
							(char *) &fc_one, sizeof(fc_one))) == -1)
			{
				ereport(LOG,
						(errcode_for_socket_access(),
				/* 翻译者：第三个 %s 是 IPv4，IPv6 或 Unix */
						 errmsg("%s(%s) failed for %s address \"%s\": %m",
								"setsockopt", "SO_REUSEADDR",
								fc_familyDesc, fc_addrDesc)));
				closesocket(fc_fd);
				continue;
			}
		}
#endif

#ifdef IPV6_V6ONLY
		if (fc_addr->ai_family == AF_INET6)
		{
			if (setsockopt(fc_fd, IPPROTO_IPV6, IPV6_V6ONLY,
						   (char *) &fc_one, sizeof(fc_one)) == -1)
			{
				ereport(LOG,
						(errcode_for_socket_access(),
				/* 翻译者：第三个 %s 是 IPv4，IPv6 或 Unix */
						 errmsg("%s(%s) failed for %s address \"%s\": %m",
								"setsockopt", "IPV6_V6ONLY",
								fc_familyDesc, fc_addrDesc)));
				closesocket(fc_fd);
				continue;
			}
		}
#endif

		/*
		 * 注意：在某些操作系统上，这可能会失败，比如旧于
		 * 2.4.21-pre3 的 Linux，它没有 IPV6_V6ONLY 套接字选项，并将
		 * ipv4 地址映射到 ipv6。它将为所有 ipv4 连接显示 ::ffff:ipv4。
		 */
		fc_err = bind(fc_fd, fc_addr->ai_addr, fc_addr->ai_addrlen);
		if (fc_err < 0)
		{
			int			fc_saved_errno = errno;

			ereport(LOG,
					(errcode_for_socket_access(),
			/* 翻译者：第一个%s是IPv4，IPv6或Unix */
					 errmsg("could not bind %s address \"%s\": %m",
							fc_familyDesc, fc_addrDesc),
					 fc_saved_errno == EADDRINUSE ?
					 (fc_addr->ai_family == AF_UNIX ?
					  errhint("Is another postmaster already running on port %d?",
							  (int) fc_portNumber) :
					  errhint("Is another postmaster already running on port %d?"
							  " If not, wait a few seconds and retry.",
							  (int) fc_portNumber)) : 0));
			closesocket(fc_fd);
			continue;
		}

#ifdef HAVE_UNIX_SOCKETS
		if (fc_addr->ai_family == AF_UNIX)
		{
			if (fc_Setup_AF_UNIX(fc_service) != STATUS_OK)
			{
				closesocket(fc_fd);
				break;
			}
		}
#endif

		/*
		 * 选择适当的接受队列长度限制。 PG_SOMAXCONN 仅
		 * 用于提供对请求的限制，适用于在请求过大时会引发内核错误的平台（有这样的情况吗？）。
		 */
		fc_maxconn = MaxBackends * 2;
		if (fc_maxconn > PG_SOMAXCONN)
			fc_maxconn = PG_SOMAXCONN;

		fc_err = listen(fc_fd, fc_maxconn);
		if (fc_err < 0)
		{
			ereport(LOG,
					(errcode_for_socket_access(),
			/* 翻译者：第一个%s是IPv4，IPv6或Unix */
					 errmsg("could not listen on %s address \"%s\": %m",
							fc_familyDesc, fc_addrDesc)));
			closesocket(fc_fd);
			continue;
		}

#ifdef HAVE_UNIX_SOCKETS
		if (fc_addr->ai_family == AF_UNIX)
			ereport(LOG,
					(errmsg("listening on Unix socket \"%s\"",
							fc_addrDesc)));
		else
#endif
			ereport(LOG,
			/* 翻译者：第一个 %s 是 IPv4 或 IPv6 */
					(errmsg("listening on %s address \"%s\", port %d",
							fc_familyDesc, fc_addrDesc, (int) fc_portNumber)));

		fc_ListenSocket[fc_listen_index] = fc_fd;
		fc_added++;
	}

	pg_freeaddrinfo_all(fc_hint.ai_family, fc_addrs);

	if (!fc_added)
		return STATUS_ERROR;

	return STATUS_OK;
}


#ifdef HAVE_UNIX_SOCKETS

/*
 * Lock_AF_UNIX -- 配置 Unix 套接字文件路径
 */
static int fc_Lock_AF_UNIX(const char *fc_unixSocketDir, const char *fc_unixSocketPath)
{
	/* 抽象套接字没有锁文件 */
	if (fc_unixSocketPath[0] == '@')
		return STATUS_OK;

	/*
	 * 获取与套接字文件相关联的互锁文件。
	 *
	 * 注意：使用套接字锁文件有两个原因，而不是直接在套接字上进行互锁。
	 * 首先，它的可移植性更强；其次，它让我们可以在没有竞争条件的情况下
	 * 删除任何预先存在的套接字文件。
	 */
	CreateSocketLockFile(fc_unixSocketPath, true, fc_unixSocketDir);

	/*
	 * 一旦我们有了互锁，我们就可以安全地删除任何预先存在的
	 * 套接字文件，以避免在 bind() 时失败。
	 */
	(void) unlink(fc_unixSocketPath);

	/*
	 * 记住套接字文件路径，以便后续维护。
	 */
	sock_paths = lappend(sock_paths, pstrdup(fc_unixSocketPath));

	return STATUS_OK;
}


/*
 * Setup_AF_UNIX -- 配置 Unix 套接字权限
 */
static int fc_Setup_AF_UNIX(const char *fc_sock_path)
{
	/* 抽象套接字没有文件系统权限 */
	if (fc_sock_path[0] == '@')
		return STATUS_OK;

	/*
	 * 如果请求，修复套接字所有权/权限。注意我们必须在
	 * listen() 之前做到这一点，以避免在此期间接受不想要的连接。
	 */
	Assert(Unix_socket_group);
	if (Unix_socket_group[0] != '\0')
	{
#ifdef WIN32
		elog(WARNING, "configuration item unix_socket_group is not supported on this platform");
#else
		char	   *fc_endptr;
		unsigned long fc_val;
		gid_t		fc_gid;

		fc_val = strtoul(Unix_socket_group, &fc_endptr, 10);
		if (*fc_endptr == '\0')
		{						/* 数字组 ID */
			fc_gid = fc_val;
		}
		else
		{						/* 将组名转换为 ID */
			struct group *fc_gr;

			fc_gr = getgrnam(Unix_socket_group);
			if (!fc_gr)
			{
				ereport(LOG,
						(errmsg("group \"%s\" does not exist",
								Unix_socket_group)));
				return STATUS_ERROR;
			}
			fc_gid = fc_gr->gr_gid;
		}
		if (chown(fc_sock_path, -1, fc_gid) == -1)
		{
			ereport(LOG,
					(errcode_for_file_access(),
					 errmsg("could not set group of file \"%s\": %m",
							fc_sock_path)));
			return STATUS_ERROR;
		}
#endif
	}

	if (chmod(fc_sock_path, Unix_socket_permissions) == -1)
	{
		ereport(LOG,
				(errcode_for_file_access(),
				 errmsg("could not set permissions of file \"%s\": %m",
						fc_sock_path)));
		return STATUS_ERROR;
	}
	return STATUS_OK;
}
#endif							/* HAVE_UNIX_SOCKETS */


/*
 * StreamConnection -- 使用
 *		服务器端口与客户端创建新的连接。将 port->sock 设置为新连接的 FD。
 *
 * 假设：这不需要非阻塞，因为
 *		Postmaster 使用 select() 来判断套接字是否准备好接受。
 *
 * 返回：STATUS_OK 或 STATUS_ERROR
 */
int StreamConnection(pgsocket fc_server_fd, Port *fc_port)
{
	/* 接受连接并填充客户端（远程）地址 */
	fc_port->raddr.salen = sizeof(fc_port->raddr.addr);
	if ((fc_port->sock = accept(fc_server_fd,
							 (struct sockaddr *) &fc_port->raddr.addr,
							 &fc_port->raddr.salen)) == PGINVALID_SOCKET)
	{
		ereport(LOG,
				(errcode_for_socket_access(),
				 errmsg("could not accept new connection: %m")));

		/*
		 * 如果 accept() 失败，则 postmaster.c 仍会将服务器
		 * 套接字视为可读的，并会立即重试。为了避免不必要地消耗大量 CPU，
		 * 在重试之前稍作延迟。
		 * （失败的最可能原因是内核文件表槽用尽；我们能做的只是希望
		 * 有些槽会被释放。）
		 */
		pg_usleep(100000L);		/* 等待 0.1 秒 */
		return STATUS_ERROR;
	}

	/* 填写服务器（本地）地址 */
	fc_port->laddr.salen = sizeof(fc_port->laddr.addr);
	if (getsockname(fc_port->sock,
					(struct sockaddr *) &fc_port->laddr.addr,
					&fc_port->laddr.salen) < 0)
	{
		ereport(LOG,
				(errmsg("%s() failed: %m", "getsockname")));
		return STATUS_ERROR;
	}

	/* 如果是 TCP 连接，请选择 NODELAY 和 KEEPALIVE 选项 */
	if (fc_port->laddr.addr.ss_family != AF_UNIX)
	{
		int			fc_on;
#ifdef WIN32
		int			oldopt;
		int			optlen;
		int			newopt;
#endif

#ifdef	TCP_NODELAY
		fc_on = 1;
		if (setsockopt(fc_port->sock, IPPROTO_TCP, TCP_NODELAY,
					   (char *) &fc_on, sizeof(fc_on)) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "setsockopt", "TCP_NODELAY")));
			return STATUS_ERROR;
		}
#endif
		fc_on = 1;
		if (setsockopt(fc_port->sock, SOL_SOCKET, SO_KEEPALIVE,
					   (char *) &fc_on, sizeof(fc_on)) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "setsockopt", "SO_KEEPALIVE")));
			return STATUS_ERROR;
		}

#ifdef WIN32

		/*
		 * 这是一个 Win32 套接字优化。 操作系统发送缓冲区的大小应该足够大，
		 * 能够一次性发送整个 Postgres 发送缓冲区，否则性能会受到影响。
		 * 如果需要发送非常大的消息，可以增大 Postgres 发送缓冲区，
		 * 但如果发生这种情况，我们不会尝试增大操作系统缓冲区，
		 * 所以要相对任意地确保 OS 缓冲区至少是 PQ_SEND_BUFFER_SIZE * 4。
		 * （当前默认值是 32kB）。
		 *
		 * 在早期的 Windows 版本中，默认的操作系统缓冲区大小是 8kB，
		 * 但在 Windows 2012 中提升到了 64kB。因此，在后来的版本中，
		 * 应该不必再更改它。不必要地更改它甚至可能会降低性能，
		 * 因为在应用程序中设置 SO_SNDBUF 会禁用在 Windows 7 中引入的
		 * “动态发送缓冲”功能。因此，在调整 SO_SNDBUF 之前，
		 * 请检查当前缓冲区大小是否已经足够大，仅在必要时增加它。
		 *
		 * 详细内容请见 https://support.microsoft.com/kb/823764/EN-US/ 和
		 * https://msdn.microsoft.com/en-us/library/bb736549%28v=vs.85%29.aspx
		 */
		optlen = sizeof(oldopt);
		if (getsockopt(fc_port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &oldopt,
					   &optlen) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "getsockopt", "SO_SNDBUF")));
			return STATUS_ERROR;
		}
		newopt = PQ_SEND_BUFFER_SIZE * 4;
		if (oldopt < newopt)
		{
			if (setsockopt(fc_port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &newopt,
						   sizeof(newopt)) < 0)
			{
				ereport(LOG,
						(errmsg("%s(%s) failed: %m", "setsockopt", "SO_SNDBUF")));
				return STATUS_ERROR;
			}
		}
#endif

		/*
		 * 还要应用当前的 keepalive 参数。如果我们设置参数失败，
		 * 则不要报错，因为这些参数并不是普遍支持的。
		 * （注意：你可能会认为在这种情况下需要将 GUC 变量重置为 0，
		 * 但这是不必要的，因为这些变量的显示钩子反正会报告真实情况。）
		 */
		(void) pq_setkeepalivesidle(tcp_keepalives_idle, fc_port);
		(void) pq_setkeepalivesinterval(tcp_keepalives_interval, fc_port);
		(void) pq_setkeepalivescount(tcp_keepalives_count, fc_port);
		(void) pq_settcpusertimeout(tcp_user_timeout, fc_port);
	}

	return STATUS_OK;
}

/*
 * StreamClose -- 关闭客户端/后端连接
 *
 * 注意：这不是用来终止会话；它只是用来释放
 * 在不再应该打开套接字的进程中的文件描述符。
 * （例如，postmaster 在将连接的所有权传递给子进程后调用此函数。）
 * 预计其他人仍然会保持套接字打开。
 * 所以，我们只想关闭描述符，不想向远端发送任何内容。
 */
void StreamClose(pgsocket fc_sock)
{
	closesocket(fc_sock);
}

/*
 * TouchSocketFiles -- 标记套接字文件为最近访问
 *
 * 该例程应该每隔一段时间调用一次，以确保套接字
 * 文件有最近的修改日期（普通的套接字操作通常不会
 * 更改修改日期）。这可以避免它们被
 * 过于热心的 /tmp 目录清理守护进程删除。
 * （我们绝不应该将套接字文件放在 /tmp 中的另一个原因…）
 */
void TouchSocketFiles(void)
{
	ListCell   *fc_l;

	/* 遍历所有创建的套接字... */
	foreach(fc_l, sock_paths)
	{
		char	   *fc_sock_path = (char *) lfirst(fc_l);

		/* 忽略错误；抱怨没有意义 */
		(void) utime(fc_sock_path, NULL);
	}
}

/*
 * RemoveSocketFiles -- 在 postmaster 关闭时解除套接字文件的链接
 */
void RemoveSocketFiles(void)
{
	ListCell   *fc_l;

	/* 遍历所有创建的套接字... */
	foreach(fc_l, sock_paths)
	{
		char	   *fc_sock_path = (char *) lfirst(fc_l);

		/* 忽略任何错误。 */
		(void) unlink(fc_sock_path);
	}
	/* 由于我们即将退出，因此无需回收存储 */
	sock_paths = NIL;
}


/* --------------------------------
 * 低级 I/O 例程从这里开始。
 *
 * 这些例程通过之前建立的连接与前端客户端进行通信。
 * --------------------------------
 */

/* --------------------------------
 * socket_set_nonblocking - 设置套接字为阻塞/非阻塞
 *
 * 如果 nonblocking 为 true，则将套接字设置为非阻塞，否则设置为阻塞。
 * --------------------------------
 */
static void fc_socket_set_nonblocking(bool fc_nonblocking)
{
	if (MyProcPort == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
				 errmsg("there is no client connection")));

	MyProcPort->noblock = fc_nonblocking;
}

/* --------------------------------
 * pq_recvbuf - 将一些字节加载到输入缓冲区
 *
 * 如果正常则返回 0，出现问题则返回 EOF
 * --------------------------------
 */
static int fc_pq_recvbuf(void)
{
	if (PqRecvPointer > 0)
	{
		if (PqRecvLength > PqRecvPointer)
		{
			/* 仍然有一些未读数据，将其左对齐到缓冲区中 */
			memmove(PqRecvBuffer, PqRecvBuffer + PqRecvPointer,
					PqRecvLength - PqRecvPointer);
			PqRecvLength -= PqRecvPointer;
			PqRecvPointer = 0;
		}
		else
			PqRecvLength = PqRecvPointer = 0;
	}

	/* 确保我们处于阻塞模式 */
	fc_socket_set_nonblocking(false);

	/* 可以从 PqRecvLength 和以上位置填充缓冲区 */
	for (;;)
	{
		int			fc_r;

		errno = 0;

		fc_r = secure_read(MyProcPort, PqRecvBuffer + PqRecvLength,
						PQ_RECV_BUFFER_SIZE - PqRecvLength);

		if (fc_r < 0)
		{
			if (errno == EINTR)
				continue;		/* 如果被中断是可以的 */

			/*
			 * 注意：尝试向客户端写入的 ereport() 会导致递归到这里，
			 * 导致堆栈溢出和核心转储！此消息必须仅发送到后端日志。
			 *
			 * 如果 errno 为零，则假设这是 EOF，让调用者来处理。
			 */
			if (errno != 0)
				ereport(COMMERROR,
						(errcode_for_socket_access(),
						 errmsg("could not receive data from client: %m")));
			return EOF;
		}
		if (fc_r == 0)
		{
			/*
			 * 检测到 EOF。我们过去在这里写入日志消息，但
			 * 现在更好的是期望最终调用者来处理这件事。
			 */
			return EOF;
		}
		/* r 包含读取的字节数，因此只需增加长度 */
		PqRecvLength += fc_r;
		return 0;
	}
}

/* --------------------------------
 * pq_getbyte - 从连接获取一个字节，或返回 EOF
 * --------------------------------
 */
int pq_getbyte(void)
{
	Assert(PqCommReadingMsg);

	while (PqRecvPointer >= PqRecvLength)
	{
		if (fc_pq_recvbuf())		/* 如果缓冲区中没有内容，则接收一些 */
			return EOF;			/* 接收数据失败 */
	}
	return (unsigned char) PqRecvBuffer[PqRecvPointer++];
}

/* --------------------------------
 * pq_peekbyte - 从连接中窥视下一个字节
 *
 * 与 pq_getbyte() 相同，只是我们不推进指针。
 * --------------------------------
 */
int pq_peekbyte(void)
{
	Assert(PqCommReadingMsg);

	while (PqRecvPointer >= PqRecvLength)
	{
		if (fc_pq_recvbuf())		/* 如果缓冲区中没有内容，则接收一些 */
			return EOF;			/* 接收数据失败 */
	}
	return (unsigned char) PqRecvBuffer[PqRecvPointer];
}

/* --------------------------------
 * pq_getbyte_if_available - 从连接获取一个字节，如果可用
 *
 * 接收到的字节存储在 *c 中。 如果读取了一个字节返回 1，
 * 如果没有数据可用返回 0，或返回 EOF 表示出现问题。
 * --------------------------------
 */
int pq_getbyte_if_available(unsigned char *fc_c)
{
	int			fc_r;

	Assert(PqCommReadingMsg);

	if (PqRecvPointer < PqRecvLength)
	{
		*fc_c = PqRecvBuffer[PqRecvPointer++];
		return 1;
	}

	/* 将套接字设置为非阻塞模式 */
	fc_socket_set_nonblocking(true);

	errno = 0;

	fc_r = secure_read(MyProcPort, fc_c, 1);
	if (fc_r < 0)
	{
		/*
		 * 如果没有数据可用而不阻塞或被中断，这是可以的（尽管
		 * 使用非阻塞套接字时，EINTR 实际上不应该发生）。报告
		 * 其他错误。
		 */
		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
			fc_r = 0;
		else
		{
			/*
			 * 注意：尝试向客户端写入的 ereport() 会导致递归到这里，
			 * 导致堆栈溢出和核心转储！此消息必须仅发送到后端日志。
			 *
			 * 如果 errno 为零，则假设这是 EOF，让调用者来处理。
			 */
			if (errno != 0)
				ereport(COMMERROR,
						(errcode_for_socket_access(),
						 errmsg("could not receive data from client: %m")));
			fc_r = EOF;
		}
	}
	else if (fc_r == 0)
	{
		/* 检测到 EOF */
		fc_r = EOF;
	}

	return fc_r;
}

/* --------------------------------
 * pq_getbytes - 从连接获取已知数量的字节
 *
 * 如果正常则返回 0，出现问题则返回 EOF
 * --------------------------------
 */
int pq_getbytes(char *fc_s, size_t fc_len)
{
	size_t		fc_amount;

	Assert(PqCommReadingMsg);

	while (fc_len > 0)
	{
		while (PqRecvPointer >= PqRecvLength)
		{
			if (fc_pq_recvbuf())	/* 如果缓冲区中没有内容，则接收一些 */
				return EOF;		/* 接收数据失败 */
		}
		fc_amount = PqRecvLength - PqRecvPointer;
		if (fc_amount > fc_len)
			fc_amount = fc_len;
		memcpy(fc_s, PqRecvBuffer + PqRecvPointer, fc_amount);
		PqRecvPointer += fc_amount;
		fc_s += fc_amount;
		fc_len -= fc_amount;
	}
	return 0;
}

/* --------------------------------
 *		pq_discardbytes		- 丢弃已知数量的字节
 *
 *		与 pq_getbytes 相同，除了我们不将数据复制到任何地方。
 *		这用于在读取错误后重新同步。
 *
 *		如果正常返回 0，如果出现问题则返回 EOF
 * --------------------------------
 */
static int fc_pq_discardbytes(size_t fc_len)
{
	size_t		fc_amount;

	Assert(PqCommReadingMsg);

	while (fc_len > 0)
	{
		while (PqRecvPointer >= PqRecvLength)
		{
			if (fc_pq_recvbuf())	/* 如果缓冲区中没有内容，则接收一些 */
				return EOF;		/* 接收数据失败 */
		}
		fc_amount = PqRecvLength - PqRecvPointer;
		if (fc_amount > fc_len)
			fc_amount = fc_len;
		PqRecvPointer += fc_amount;
		fc_len -= fc_amount;
	}
	return 0;
}

/* --------------------------------
 *		pq_buffer_has_data		- 是否有任何缓冲数据可供读取？
 *
 * 这不会尝试读取更多数据。
 * --------------------------------
 */
bool pq_buffer_has_data(void)
{
	return (PqRecvPointer < PqRecvLength);
}


/* --------------------------------
 *		pq_startmsgread - 开始从客户端读取消息。
 *
 *		这必须在调用任何 pq_get* 函数之前。
 * --------------------------------
 */
void pq_startmsgread(void)
{
	/*
	 * 不应该已经有活动的读取，但我们还是检查一下
	 * 确保。
	 */
	if (PqCommReadingMsg)
		ereport(FATAL,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("terminating connection because protocol synchronization was lost")));

	PqCommReadingMsg = true;
}


/* --------------------------------
 *		pq_endmsgread	- 完成读取消息。
 *
 *		这必须在使用 pq_getbytes() 和相关函数读取消息后调用，
 *		以表明我们已经读取了整个消息。
 *		pq_getmessage() 隐式执行此操作。
 * --------------------------------
 */
void pq_endmsgread(void)
{
	Assert(PqCommReadingMsg);

	PqCommReadingMsg = false;
}

/* --------------------------------
 *		pq_is_reading_msg - 我们当前正在读取消息吗？
 *
 * 这用于外部空闲循环中的错误恢复，以检测我们是否
 * 丢失了协议同步，并需要终止连接。 pq_startmsgread()
 * 也会检查这个，但提前检测会更好。
 * --------------------------------
 */
bool pq_is_reading_msg(void)
{
	return PqCommReadingMsg;
}

/* --------------------------------
 *		pq_getmessage	- 从连接获取带长度字的消息
 *
 *		返回值放入一个可扩展的 StringInfo，该 StringInfo 已
 *		被调用者初始化。
 *		只有消息体被放入 StringInfo；长度字被移除。
 *		此外，s->cursor 被初始化为零，以方便扫描消息内容。
 *
 *		maxlen 是我们愿意接受的
 *		消息长度的上限。如果客户端试图发送超过这个长度的消息，
 *		我们将中止连接（通过返回 EOF）。
 *
 *		如果正常返回 0，如果出现问题则返回 EOF
 * --------------------------------
 */
int pq_getmessage(StringInfo fc_s, int fc_maxlen)
{
	int32		fc_len;

	Assert(PqCommReadingMsg);

	resetStringInfo(fc_s);

	/* 读取消息长度字 */
	if (pq_getbytes((char *) &fc_len, 4) == EOF)
	{
		ereport(COMMERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("unexpected EOF within message length word")));
		return EOF;
	}

	fc_len = pg_ntoh32(fc_len);

	if (fc_len < 4 || fc_len > fc_maxlen)
	{
		ereport(COMMERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("invalid message length")));
		return EOF;
	}

	fc_len -= 4;					/* 折扣长度本身 */

	if (fc_len > 0)
	{
		/*
		 * 为消息分配空间。如果我们没有足够的空间（异常
		 * 大的消息），我们将 elog(ERROR)，但我们想要丢弃
		 * 消息体，以避免失去通信同步。
		 */
		PG_TRY();
		{
			enlargeStringInfo(fc_s, fc_len);
		}
		PG_CATCH();
		{
			if (fc_pq_discardbytes(fc_len) == EOF)
				ereport(COMMERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg("incomplete message from client")));

			/* 我们丢弃了消息的其余部分，因此我们恢复了同步。 */
			PqCommReadingMsg = false;
			PG_RE_THROW();
		}
		PG_END_TRY();

		/* 并获取消息 */
		if (pq_getbytes(fc_s->data, fc_len) == EOF)
		{
			ereport(COMMERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("incomplete message from client")));
			return EOF;
		}
		fc_s->len = fc_len;
		/* 根据 StringInfo 约定放置尾随的 null */
		fc_s->data[fc_len] = '\0';
	}

	/* 完成读取消息。 */
	PqCommReadingMsg = false;

	return 0;
}


static int fc_internal_putbytes(const char *fc_s, size_t fc_len)
{
	size_t		fc_amount;

	while (fc_len > 0)
	{
		/* 如果缓冲区满了，则将其刷新 */
		if (PqSendPointer >= PqSendBufferSize)
		{
			fc_socket_set_nonblocking(false);
			if (fc_internal_flush())
				return EOF;
		}
		fc_amount = PqSendBufferSize - PqSendPointer;
		if (fc_amount > fc_len)
			fc_amount = fc_len;
		memcpy(PqSendBuffer + PqSendPointer, fc_s, fc_amount);
		PqSendPointer += fc_amount;
		fc_s += fc_amount;
		fc_len -= fc_amount;
	}
	return 0;
}

/* --------------------------------
 *		socket_flush		- 刷新待处理的输出
 *
 *		如果正常返回 0，如果出现问题则返回 EOF
 * --------------------------------
 */
static int fc_socket_flush(void)
{
	int			fc_res;

	/* 如果是重入调用则不操作 */
	if (PqCommBusy)
		return 0;
	PqCommBusy = true;
	fc_socket_set_nonblocking(false);
	fc_res = fc_internal_flush();
	PqCommBusy = false;
	return fc_res;
}

/* --------------------------------
 *		internal_flush - 刷新待处理输出
 *
 * 返回0表示正常（意味着所有内容都已发送，或操作将阻塞并且套接字处于非阻塞模式），如果有问题则返回EOF。
 * --------------------------------
 */
static int fc_internal_flush(void)
{
	static int	fc_last_reported_send_errno = 0;

	char	   *fc_bufptr = PqSendBuffer + PqSendStart;
	char	   *fc_bufend = PqSendBuffer + PqSendPointer;

	while (fc_bufptr < fc_bufend)
	{
		int			fc_r;

		fc_r = secure_write(MyProcPort, fc_bufptr, fc_bufend - fc_bufptr);

		if (fc_r <= 0)
		{
			if (errno == EINTR)
				continue;		/* 如果我们被中断，则正常 */

			/*
			 * 如果没有数据可在不阻塞的情况下写入，并且套接字处于
			 * 非阻塞模式下，则正常。
			 */
			if (errno == EAGAIN ||
				errno == EWOULDBLOCK)
			{
				return 0;
			}

			/*
			 * 小心：一个尝试向客户端写入的ereport()会导致递归到此，
			 * 导致堆栈溢出和核心转储！该消息必须仅发送到主进程日志。
			 *
			 * 如果在输出过程中客户端断开连接，我们可能会在
			 * 到达安全查询中止点之前写入相当多的数据。
			 * 所以，抑制重复的日志消息。
			 */
			if (errno != fc_last_reported_send_errno)
			{
				fc_last_reported_send_errno = errno;
				ereport(COMMERROR,
						(errcode_for_socket_access(),
						 errmsg("could not send data to client: %m")));
			}

			/*
			 * 我们无论如何都会丢掉缓冲数据，以便处理可以
			 * 继续，即使我们可能很快就会退出。我们还设置一个
			 * 标志，它将导致下一个CHECK_FOR_INTERRUPTS终止
			 * 连接。
			 */
			PqSendStart = PqSendPointer = 0;
			ClientConnectionLost = 1;
			InterruptPending = 1;
			return EOF;
		}

		fc_last_reported_send_errno = 0;	/* 在任何成功发送后重置 */
		fc_bufptr += fc_r;
		PqSendStart += fc_r;
	}

	PqSendStart = PqSendPointer = 0;
	return 0;
}

/* --------------------------------
 *		pq_flush_if_writable - 如果可在不阻塞的情况下写入，则刷新待处理输出
 *
 * 返回0表示正常，或如果有问题则返回EOF。
 * --------------------------------
 */
static int fc_socket_flush_if_writable(void)
{
	int			fc_res;

	/* 如果没有事情可做，则快速退出 */
	if (PqSendPointer == PqSendStart)
		return 0;

	/* 如果是重入调用则不操作 */
	if (PqCommBusy)
		return 0;

	/* 暂时将套接字置于非阻塞模式 */
	fc_socket_set_nonblocking(true);

	PqCommBusy = true;
	fc_res = fc_internal_flush();
	PqCommBusy = false;
	return fc_res;
}

/* --------------------------------
 *	socket_is_send_pending - 输出缓冲区中是否有待处理数据？
 * --------------------------------
 */
static bool fc_socket_is_send_pending(void)
{
	return (PqSendStart < PqSendPointer);
}

/* --------------------------------
 * 消息级I/O例程从这里开始。
 * --------------------------------
 */


/* --------------------------------
 *		socket_putmessage - 发送正常消息（在COPY OUT模式下被抑制）
 *
 *		msgtype是要放在消息体之前的消息类型代码。
 *
 *		len是*s中消息体数据的长度。该例程会插入一个消息长度
 *		字（等于len+4，因为它也计算了自身）。
 *
 *		我们会抑制在pqcomm.c繁忙时生成的消息。这
 *		避免了消息被插入到其他消息中的任何可能性。唯一已知的
 *		问题是在pqcomm.c例程期间发生SIGQUIT --- quickdie()会尝试发送警告
 *		消息，而最合理的方法似乎是丢弃它。
 *
 *		返回0表示正常，EOF表示有问题
 * --------------------------------
 */
static int fc_socket_putmessage(char fc_msgtype, const char *fc_s, size_t fc_len)
{
	uint32		fc_n32;

	Assert(fc_msgtype != 0);

	if (PqCommBusy)
		return 0;
	PqCommBusy = true;
	if (fc_internal_putbytes(&fc_msgtype, 1))
		goto fail;

	fc_n32 = pg_hton32((uint32) (fc_len + 4));
	if (fc_internal_putbytes((char *) &fc_n32, 4))
		goto fail;

	if (fc_internal_putbytes(fc_s, fc_len))
		goto fail;
	PqCommBusy = false;
	return 0;

fail:
	PqCommBusy = false;
	return EOF;
}

/* --------------------------------
 *		pq_putmessage_noblock - 像pq_putmessage，但绝不阻塞
 *
 *		如果输出缓冲区太小，无法容纳消息，则该缓冲区
 *		会被扩大。
 */
static void fc_socket_putmessage_noblock(char fc_msgtype, const char *fc_s, size_t fc_len)
{
	int			fc_res PG_USED_FOR_ASSERTS_ONLY;
	int			fc_required;

	/*
	 * 确保输出缓冲区有足够的空间供消息头和
	 * 消息本身使用。
	 */
	fc_required = PqSendPointer + 1 + 4 + fc_len;
	if (fc_required > PqSendBufferSize)
	{
		PqSendBuffer = repalloc(PqSendBuffer, fc_required);
		PqSendBufferSize = fc_required;
	}
	fc_res = pq_putmessage(fc_msgtype, fc_s, fc_len);
	Assert(fc_res == 0);			/* 当消息适合缓冲区时不应失败 */
}

/* --------------------------------
 *        pq_putmessage_v2 - 发送协议版本 2 的消息
 *
 *        msgtype 是一个消息类型代码，用于放置在消息正文之前。
 *
 *        我们不再支持协议版本 2，但我们保留这个
 *        函数，以便如果客户端尝试使用协议版本 2 连接，
 *        我们可以礼貌地仍然以旧格式向客户端发送
 *        “不支持的协议版本”错误。
 *
 *        和 pq_putmessage() 中一样，我们在
 *        pqcomm.c 繁忙时抑制生成的消息。
 *
 *        如果正常返回 0，如果有问题返回 EOF
 * --------------------------------
 */
int pq_putmessage_v2(char fc_msgtype, const char *fc_s, size_t fc_len)
{
	Assert(fc_msgtype != 0);

	if (PqCommBusy)
		return 0;
	PqCommBusy = true;
	if (fc_internal_putbytes(&fc_msgtype, 1))
		goto fail;

	if (fc_internal_putbytes(fc_s, fc_len))
		goto fail;
	PqCommBusy = false;
	return 0;

fail:
	PqCommBusy = false;
	return EOF;
}

/*
 * TCP Keepalive 参数的支持
 */

/*
 * 在 Windows 上，我们需要同时设置空闲和间隔。
 * 我们也无法将它们重置为默认值（设置为零将
 * 实际上将它们设置为零，而不是默认值），因此我们回退到
 * 开箱即用的默认值。
 */
#if defined(WIN32) && defined(SIO_KEEPALIVE_VALS)
static int pq_setkeepaliveswin32(Port *port, int idle, int interval)
{
	struct tcp_keepalive ka;
	DWORD		retsize;

	if (idle <= 0)
		idle = 2 * 60 * 60;		/* 默认 = 2 小时 */
	if (interval <= 0)
		interval = 1;			/* 默认 = 1 秒 */

	ka.onoff = 1;
	ka.keepalivetime = idle * 1000;
	ka.keepaliveinterval = interval * 1000;

	if (WSAIoctl(port->sock,
				 SIO_KEEPALIVE_VALS,
				 (LPVOID) &ka,
				 sizeof(ka),
				 NULL,
				 0,
				 &retsize,
				 NULL,
				 NULL)
		!= 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) failed: error code %d",
						"WSAIoctl", "SIO_KEEPALIVE_VALS", WSAGetLastError())));
		return STATUS_ERROR;
	}
	if (port->keepalives_idle != idle)
		port->keepalives_idle = idle;
	if (port->keepalives_interval != interval)
		port->keepalives_interval = interval;
	return STATUS_OK;
}
#endif

int pq_getkeepalivesidle(Port *fc_port)
{
#if defined(PG_TCP_KEEPALIVE_IDLE) || defined(SIO_KEEPALIVE_VALS)
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return 0;

	if (fc_port->keepalives_idle != 0)
		return fc_port->keepalives_idle;

	if (fc_port->default_keepalives_idle == 0)
	{
#ifndef WIN32
		socklen_t	fc_size = sizeof(fc_port->default_keepalives_idle);

		if (getsockopt(fc_port->sock, IPPROTO_TCP, PG_TCP_KEEPALIVE_IDLE,
					   (char *) &fc_port->default_keepalives_idle,
					   &fc_size) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "getsockopt", PG_TCP_KEEPALIVE_IDLE_STR)));
			fc_port->default_keepalives_idle = -1; /* 不知道 */
		}
#else							/* WIN32 */
		/* 我们无法在 Windows 上获取默认值，因此返回“不知道” */
		fc_port->default_keepalives_idle = -1;
#endif							/* WIN32 */
	}

	return fc_port->default_keepalives_idle;
#else
	return 0;
#endif
}

int pq_setkeepalivesidle(int fc_idle, Port *fc_port)
{
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return STATUS_OK;

/* 在这里检查 SIO_KEEPALIVE_VALS，而不仅仅是 WIN32，因为某些工具链缺少它 */
#if defined(PG_TCP_KEEPALIVE_IDLE) || defined(SIO_KEEPALIVE_VALS)
	if (fc_idle == fc_port->keepalives_idle)
		return STATUS_OK;

#ifndef WIN32
	if (fc_port->default_keepalives_idle <= 0)
	{
		if (pq_getkeepalivesidle(fc_port) < 0)
		{
			if (fc_idle == 0)
				return STATUS_OK;	/* 默认已设置但未知 */
			else
				return STATUS_ERROR;
		}
	}

	if (fc_idle == 0)
		fc_idle = fc_port->default_keepalives_idle;

	if (setsockopt(fc_port->sock, IPPROTO_TCP, PG_TCP_KEEPALIVE_IDLE,
				   (char *) &fc_idle, sizeof(fc_idle)) < 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) failed: %m", "setsockopt", PG_TCP_KEEPALIVE_IDLE_STR)));
		return STATUS_ERROR;
	}

	fc_port->keepalives_idle = fc_idle;
#else							/* WIN32 */
	return pq_setkeepaliveswin32(fc_port, idle, fc_port->keepalives_interval);
#endif
#else
	if (idle != 0)
	{
		ereport(LOG,
				(errmsg("setting the keepalive idle time is not supported")));
		return STATUS_ERROR;
	}
#endif

	return STATUS_OK;
}

int pq_getkeepalivesinterval(Port *fc_port)
{
#if defined(TCP_KEEPINTVL) || defined(SIO_KEEPALIVE_VALS)
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return 0;

	if (fc_port->keepalives_interval != 0)
		return fc_port->keepalives_interval;

	if (fc_port->default_keepalives_interval == 0)
	{
#ifndef WIN32
		socklen_t	fc_size = sizeof(fc_port->default_keepalives_interval);

		if (getsockopt(fc_port->sock, IPPROTO_TCP, TCP_KEEPINTVL,
					   (char *) &fc_port->default_keepalives_interval,
					   &fc_size) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "getsockopt", "TCP_KEEPINTVL")));
			fc_port->default_keepalives_interval = -1; /* 不知道 */
		}
#else
		/* 我们无法在 Windows 上获取默认值，因此返回“不知道” */
		fc_port->default_keepalives_interval = -1;
#endif							/* WIN32 */
	}

	return fc_port->default_keepalives_interval;
#else
	return 0;
#endif
}

int pq_setkeepalivesinterval(int fc_interval, Port *fc_port)
{
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return STATUS_OK;

#if defined(TCP_KEEPINTVL) || defined(SIO_KEEPALIVE_VALS)
	if (fc_interval == fc_port->keepalives_interval)
		return STATUS_OK;

#ifndef WIN32
	if (fc_port->default_keepalives_interval <= 0)
	{
		if (pq_getkeepalivesinterval(fc_port) < 0)
		{
			if (fc_interval == 0)
				return STATUS_OK;	/* 默认已设置但未知 */
			else
				return STATUS_ERROR;
		}
	}

	if (fc_interval == 0)
		fc_interval = fc_port->default_keepalives_interval;

	if (setsockopt(fc_port->sock, IPPROTO_TCP, TCP_KEEPINTVL,
				   (char *) &fc_interval, sizeof(fc_interval)) < 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) failed: %m", "setsockopt", "TCP_KEEPINTVL")));
		return STATUS_ERROR;
	}

	fc_port->keepalives_interval = fc_interval;
#else							/* WIN32 */
	return pq_setkeepaliveswin32(fc_port, fc_port->keepalives_idle, fc_interval);
#endif
#else
	if (fc_interval != 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) not supported", "setsockopt", "TCP_KEEPINTVL")));
		return STATUS_ERROR;
	}
#endif

	return STATUS_OK;
}

int pq_getkeepalivescount(Port *fc_port)
{
#ifdef TCP_KEEPCNT
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return 0;

	if (fc_port->keepalives_count != 0)
		return fc_port->keepalives_count;

	if (fc_port->default_keepalives_count == 0)
	{
		socklen_t	fc_size = sizeof(fc_port->default_keepalives_count);

		if (getsockopt(fc_port->sock, IPPROTO_TCP, TCP_KEEPCNT,
					   (char *) &fc_port->default_keepalives_count,
					   &fc_size) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "getsockopt", "TCP_KEEPCNT")));
			fc_port->default_keepalives_count = -1;	/* 不知道 */
		}
	}

	return fc_port->default_keepalives_count;
#else
	return 0;
#endif
}

int pq_setkeepalivescount(int fc_count, Port *fc_port)
{
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return STATUS_OK;

#ifdef TCP_KEEPCNT
	if (fc_count == fc_port->keepalives_count)
		return STATUS_OK;

	if (fc_port->default_keepalives_count <= 0)
	{
		if (pq_getkeepalivescount(fc_port) < 0)
		{
			if (fc_count == 0)
				return STATUS_OK;	/* 默认已设置但未知 */
			else
				return STATUS_ERROR;
		}
	}

	if (fc_count == 0)
		fc_count = fc_port->default_keepalives_count;

	if (setsockopt(fc_port->sock, IPPROTO_TCP, TCP_KEEPCNT,
				   (char *) &fc_count, sizeof(fc_count)) < 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) failed: %m", "setsockopt", "TCP_KEEPCNT")));
		return STATUS_ERROR;
	}

	fc_port->keepalives_count = fc_count;
#else
	if (fc_count != 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) not supported", "setsockopt", "TCP_KEEPCNT")));
		return STATUS_ERROR;
	}
#endif

	return STATUS_OK;
}

int pq_gettcpusertimeout(Port *fc_port)
{
#ifdef TCP_USER_TIMEOUT
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return 0;

	if (fc_port->tcp_user_timeout != 0)
		return fc_port->tcp_user_timeout;

	if (fc_port->default_tcp_user_timeout == 0)
	{
		socklen_t	fc_size = sizeof(fc_port->default_tcp_user_timeout);

		if (getsockopt(fc_port->sock, IPPROTO_TCP, TCP_USER_TIMEOUT,
					   (char *) &fc_port->default_tcp_user_timeout,
					   &fc_size) < 0)
		{
			ereport(LOG,
					(errmsg("%s(%s) failed: %m", "getsockopt", "TCP_USER_TIMEOUT")));
			fc_port->default_tcp_user_timeout = -1;	/* 不知道 */
		}
	}

	return fc_port->default_tcp_user_timeout;
#else
	return 0;
#endif
}

int pq_settcpusertimeout(int fc_timeout, Port *fc_port)
{
	if (fc_port == NULL || fc_port->laddr.addr.ss_family == AF_UNIX)
		return STATUS_OK;

#ifdef TCP_USER_TIMEOUT
	if (fc_timeout == fc_port->tcp_user_timeout)
		return STATUS_OK;

	if (fc_port->default_tcp_user_timeout <= 0)
	{
		if (pq_gettcpusertimeout(fc_port) < 0)
		{
			if (fc_timeout == 0)
				return STATUS_OK;	/* 默认已设置但未知 */
			else
				return STATUS_ERROR;
		}
	}

	if (fc_timeout == 0)
		fc_timeout = fc_port->default_tcp_user_timeout;

	if (setsockopt(fc_port->sock, IPPROTO_TCP, TCP_USER_TIMEOUT,
				   (char *) &fc_timeout, sizeof(fc_timeout)) < 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) failed: %m", "setsockopt", "TCP_USER_TIMEOUT")));
		return STATUS_ERROR;
	}

	fc_port->tcp_user_timeout = fc_timeout;
#else
	if (fc_timeout != 0)
	{
		ereport(LOG,
				(errmsg("%s(%s) not supported", "setsockopt", "TCP_USER_TIMEOUT")));
		return STATUS_ERROR;
	}
#endif

	return STATUS_OK;
}

/*
 * 检查客户端是否仍然连接。
 */
bool pq_check_connection(void)
{
	WaitEvent	fc_events[FeBeWaitSetNEvents];
	int			fc_rc;

	/*
	 * 修改套接字事件过滤器而不恢复是可以的，因为
	 * 所有 FeBeWaitSet 套接字等待站点都是这样做的。
	 */
	ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, WL_SOCKET_CLOSED, NULL);

retry:
	fc_rc = WaitEventSetWait(FeBeWaitSet, 0, fc_events, lengthof(fc_events), 0);
	for (int fc_i = 0; fc_i < fc_rc; ++fc_i)
	{
		if (fc_events[fc_i].events & WL_SOCKET_CLOSED)
			return false;
		if (fc_events[fc_i].events & WL_LATCH_SET)
		{
			/*
			 * 一个闩锁事件可能会阻止其他事件被
			 * 报告。重置它并重新轮询。无需恢复它
			 * 因为没有代码应该期望闩锁能跨越
			 * CHECK_FOR_INTERRUPTS() 存在。
			 */
			ResetLatch(MyLatch);
			goto retry;
		}
	}

	return true;
}
